# Pyro based server
import math
import Pyro.core
import Pyro.naming
import Pyro.protocol
import sys
import threading
import traceback
import hashlib
import time
import copy
import os

from ubiqis.util import get_internal_ip as getip

HOST_RADIUS = 3

class Mergeable:
    def __init__(self,value):
        self.value = value
    def __str__(self):
        return str(self.value)
    def merge(self,value):
        if isinstance(value,Mergeable):
            value = value.value
        if self.value == None:
            self.value = value
        elif type(self.value) == set:
            if type(value) == set:
                for s in value:
                    self.value.add(s)
            else:
                self.value.add(value)

def dump_exc(e):
    traceback.print_stack()
    print "Message:",e,type(e)

def shahex(x):
    m = hashlib.sha1()
    m.update(x)
    h = m.hexdigest()
    return int(h,16)

class Ubiserv(Pyro.core.ObjBase):
    """
    Peer to peer Pyro based server
    see http://pyro.sourceforge.net/
    Keys to success:
    1) No calls to Pyro objects while a lock is held
    2) No read/write of self.known / self.data without lock. E.g. loops
       that contain pyro calls should be based on copies of self.known
       or self.data, to avoid mutating keys.
    3) Minimize number of pyro calls on stack frame (avoid
       networked recursion).
    """
    def __init__(self,daemon,sharedir='/var/cache/ubiqis'):
        Pyro.core.ObjBase.__init__(self)
        self.known = {}
        self.pri = {}
        self.data = {}
        self.ip = getip()
        self.rawuri = daemon.connect(self,"ubiqis")
        self.port = daemon.port
        self.uri = "PYROLOC://"+self.ip+":"+str(self.port)+"/ubiqis"
        self.shahexv = shahex(self.uri)
        self.known[self.uri] = 0
        self.lock = threading.RLock()
        self.dir = sharedir

    def mkhash(self,x):
        return self.shortpath(shahex(x)-self.shahexv)

    def shortpath(self,x):
        big = (1<<159)
        if abs(x-big) < abs(x):
            return x-big
        elif abs(x+big) <= abs(x):
            return x+big
        else:
            return x

    def update_pri(self):
        """
        Keep track of whether another host is a hi/lo priority
        server. If a host is 'hi' priority, we will pass on
        store() and hello() information. Otherwise, we will
        not. Hosts are stored in a ring that is 1<<160 bits around.
        Nearest HOST_RADIUS hosts are kept in high on each side to
        maintain ring integrity. One host in each k-bucket is
        kept in hi for fast lookup.
        """
        def dist(x):
            return abs(self.known[x])
        s = sorted(self.known,key=dist)
        pri = {}
        for k in s:
            self.pri[k] = 'lo'
        last_bucket = -1
        n = 0
        mask = 0
        cw = []
        ccw = []
        for k in s:
            v = self.known[k]
            if v > 0 and len(ccw) < HOST_RADIUS:
                ccw.append(k)
            if v < 0 and len(cw) < HOST_RADIUS:
                cw.append(k)
            v = abs(v)
            p = 'lo'
            if (v & mask) != v:
                n = n + 1
                mask = (1<<n)-1
                bucket = n / 5
                if bucket != last_bucket:
                    last_bucket = bucket
                    p = 'hi'
            pri[k] = p
        for k in cw:
            pri[k] = 'h+' # h+ is higher than hi
        for k in ccw:
            pri[k] = 'h+'
        pri[self.uri] = 'h+'
        self.pri = pri

    def hello(self,others):
        try:
            self.lock.acquire()
            for other in others:
                if not self.known.has_key(other):
                    self.known[other] = self.mkhash(other)
            self.update_pri()
            return [h for h in self.pri if self.pri[h] != 'lo']
        except Exception, e:
            dump_exc(e)
        finally:
            self.lock.release()

    def make_friends(self):
        """
        Very basic implementation. Could be improved and optimized
        in a lot of ways.
        """
        def dist(x):
            return abs(self.known[x])
        def byk(x):
            return self.known[x]
        try:
            self.lock.acquire()
            hosts = sorted(self.known,key=byk)
        finally:
            self.lock.release()
        while(True):
            try:
                self.lock.acquire()
                hosts = sorted(self.known,key=byk)
                if len(hosts) > 0 and hosts[0] < 0 and hosts[-1] > 0:
                    return
                new_hosts = []
                shosts = sorted(self.known,key=dist)
            finally:
                self.lock.release()
            for host in shosts:
                if host == self.uri:
                    continue
                try:
                    po = Pyro.core.getProxyForURI(host)
                    answer = po.hello([self.uri])
                    for ahost in answer:
                        new_hosts.append(ahost)
                except Exception, e:
                    dump_exc(e)
            try:
                self.lock.acquire()
                shosts = []
                for host in new_hosts:
                    if host not in self.known:
                        self.known[host] = self.mkhash(host)
                        shosts.append(host)
                self.update_pri()
            finally:
                self.lock.release()
            if len(shosts) == 0:
                return

    def nearest(self,key,requestor):
        try:
            self.lock.acquire()
            self.add_new([requestor])
            v = self.mkhash(key)
            def dist(k):
                return abs(self.shortpath(self.known[k]-v))
            d = dist(requestor)
            hosts = [h for h in self.pri if self.pri[h] != 'lo']
            hosts = sorted(hosts,key=dist)
            near_hosts = []
            for host in hosts:
                dd = dist(host)
                if dist(host) < d:
                    near_hosts.append(host)
                if len(near_hosts) >= 2*HOST_RADIUS:
                    break
            return near_hosts
        except Exception, e:
            dump_exc(e)
            return []
        finally:
            self.lock.release()

    def store(self,key,value):
        #print self.uri,'.store(',key,')'
        try:
            self.make_friends()
            hosts = []
            known = {}
            v = self.mkhash(key)
            def dist(k):
                return abs(self.shortpath(known[k]-v))
            try:
                self.lock.acquire()
                hosts = [h for h in self.pri if self.pri[h] != 'lo']
                for h in hosts:
                    known[h] = self.known[h]
                hosts = sorted(hosts,key=dist)
            except Exception, e:
                dump_exc(e)
            finally:
                self.lock.release()
            while len(hosts) > 0:
                dead = set()
                for h in hosts:
                    if h == self.uri:
                        continue
                    new_hosts = []
                    try:
                        po = Pyro.core.getProxyForURI(h)
                        phosts = po.nearest(key,self.uri)
                        for ph in phosts:
                            if ph not in known:
                                known[ph] = self.mkhash(ph)
                                if dist(ph) < dist(hosts[0]):
                                    new_hosts.append(ph)
                        hosts = sorted(new_hosts,key=dist)
                        self.add_new(hosts)
                        break
                    except Exception, e:
                        dump_exc(e)
                        dead.add(hosts)
                        del known[h]
                self.clear_dead(dead)
            hosts = sorted(known,key=dist)
            dead = set()
            for i in xrange(0,min(len(hosts),HOST_RADIUS)):
                if hosts[i] == self.uri:
                    self._store(key,value)
                else:
                    try:
                        po = Pyro.core.getProxyForURI(hosts[i])
                        po._store(key,value)
                    except Exception, e:
                        dump_exc(e)
                        dead.add(host)
            self.add_new(hosts)
            self.clear_dead(dead)
        except Exception,e:
            dump_exc(e)

    def query(self,key):
        #print self.uri,'.query(',key,')'
        try:
            self.make_friends()
            hosts = []
            known = {}
            v = self.mkhash(key)
            def dist(k):
                return abs(self.shortpath(known[k]-v))
            try:
                self.lock.acquire()
                hosts = [h for h in self.pri if self.pri[h] != 'lo']
                for h in hosts:
                    known[h] = self.known[h]
                hosts = sorted(hosts,key=dist)
            except Exception, e:
                dump_exc(e)
            finally:
                self.lock.release()
            while len(hosts) > 0:
                dead = set()
                for h in hosts:
                    if h == self.uri:
                        continue
                    new_hosts = []
                    try:
                        po = Pyro.core.getProxyForURI(h)
                        phosts = po.nearest(key,self.uri)
                        for ph in phosts:
                            if ph not in known:
                                known[ph] = self.mkhash(ph)
                                if dist(ph) < dist(hosts[0]):
                                    new_hosts.append(ph)
                        hosts = sorted(new_hosts,key=dist)
                        self.add_new(hosts)
                        break
                    except Exception, e:
                        dump_exc(e)
                        dead.add(hosts)
                        del known[h]
                self.clear_dead(dead)
            hosts = sorted(known,key=dist)
            dead = set()
            answer = None
            for i in xrange(0,min(len(hosts),HOST_RADIUS)):
                if hosts[i] == self.uri:
                    answer = self._query(key)
                    if answer != None:
                        break
                else:
                    try:
                        po = Pyro.core.getProxyForURI(hosts[i])
                        answer = po._query(key)
                        if answer != None:
                            break
                    except Exception, e:
                        dump_exc(e)
                        dead.add(host)
            self.add_new(hosts)
            self.clear_dead(dead)
            return answer
        except Exception,e:
            dump_exc(e)
            return None

    def _store(self,key,value):
        #print self.uri,'._store(',key,')'
        try:
            self.lock.acquire()
            if self.data.has_key(key) and isinstance(self.data[key],Mergeable):
                self.data[key].merge(value)
            else:
                self.data[key] = value
        finally:
            self.lock.release()

    def clear_dead(self,dead):
        try:
            self.lock.acquire()
            for host in dead:
                del self.known[host]
            self.update_pri()
        finally:
            self.lock.release()

    def add_new(self,hosts):
        try:
            self.lock.acquire()
            if hosts == None:
                print 'hosts was none'
                return
            new_hosts = set()
            for host in hosts:
                if host not in self.known:
                    new_hosts.add(host)
            if len(new_hosts) > 0:
                for host in new_hosts:
                    self.known[host] = self.mkhash(host)
                self.update_pri()
        except Exception, e:
            dump_exc(e)
        finally:
            self.lock.release()

    def _query(self,key):
        try:
            self.lock.acquire()
            if self.data.has_key(key):
                return (self.data[key],self.uri)
            else:
                return (None,None)
        finally:
            self.lock.release()

    def ping(self):
        pass

    def share_file(self,file):
        """
        Either a relative or absolute path to a file in the
        shared directory.
        """
        full = file
        if not os.path.exists(file):
            full = os.path.join(self.dir,file)
        full = os.path.abspath(full)
        if not os.path.exists(full):
            raise Exception(file+" does not exist")
        if not full.startswith(self.dir):
            raise Exception(file+" is not in shared directory")
        full = full[len(self.dir)+1:]
        s = set()
        s.add(self.uri)
        m = Mergeable(s)
        self.store(full,m)

    def read(self,file,seek,bytes):
        """
        Read bytes from a remote file and return them. The file
        must be in the shared directory.
        """
        full = file
        if not os.path.exists(file):
            full = os.path.join(self.dir,file)
        full = os.path.abspath(full)
        if not os.path.exists(full):
            raise IOError(file+" does not exist")
        if not full.startswith(self.dir):
            raise IOError(file+" is not in shared directory")
        fd = open(full,"r")
        try:
            fd.seek(seek)
            return fd.read(bytes)
        finally:
            fd.close()

    def read_network(self,net,key=None):
        """
        Prints the entire network. This is useful for debugging only.
        """
        if self.uri in net:
            return net
        try:
            self.lock.acquire()
            net[self.uri] = {}
            net[self.uri]['known'] = copy.deepcopy(self.pri)
            net[self.uri]['data'] = copy.deepcopy(self.data)
            if key != None:
                net[self.uri]['key'] = float(self.mkhash(key)/(1.0*(1<<160)))
        finally:
            self.lock.release()
        for host in net[self.uri]['known']:
            try:
                po = Pyro.core.getProxyForURI(host)
                new_net = po.read_network(net,key)
                net = new_net
            except Exception, e:
                dump_exc(e)
        return net

def copy_remote_file(poc,file,dest,blocksize=1024):
    """
    Search a local node for the owner of a file. Then
    read the file from the node, copying it to the
    destination.
    """
    (m,s) = poc.query(file)
    if m == None or not isinstance(m,Mergeable):
        raise IOError(file+" not found on network")
    fw = open(dest,"w")
    po = None
    for p in m.value:
        try:
            print '[',p,']'
            po = Pyro.core.getProxyForURI(p)
            break
        except Exception, e:
            #dump_exc(e)
            pass
    seek = 0
    while True:
        buf = po.read(file,seek,blocksize)
        if buf == '':
            break
        seek = seek + len(buf)
        fw.write(buf)
    fw.close()

def main(argv):
    Pyro.core.initServer()
    daemon = Pyro.core.Daemon(port=int(argv[1]))
    u = Ubiserv(daemon,"/var/cache/ubiqis")
    uri = u.uri
    print 'port',argv[1]

    for host in argv[2:]:
        while True:
            try:
                print 'hello to',host
                u.hello([host])
                #po = Pyro.core.getProxyForURI(host)
                #po._setOneway("hello")
                #po.hello(uri)
                u.make_friends()
                break
            except Exception, e:
                dump_exc(e)
                time.sleep(1)
        print '  connected to',host

    print "running",uri

    daemon.requestLoop()

if __name__ == '__main__':
    main(sys.argv)
